package cn.ztuo.bitrade.consumer;

import cn.ztuo.bitrade.Trade.CoinTraderFactory;
import cn.ztuo.bitrade.entity.ContractCoin;
import cn.ztuo.bitrade.runnable.RobotOrderRunnable;
import cn.ztuo.bitrade.service.ContractService;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

@Slf4j
@Component
public class ContractRobotConsumer {
    @Autowired
    private ContractService contractService;


    @KafkaListener(topics = "contract-robot", containerFactory = "kafkaListenerContainerFactory")
    public void onAddCoinTradeByContractCoin(List<ConsumerRecord<String, String>> records) {
        Map<String, Integer> map = RobotOrderRunnable.robotStatusMap;
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String, String> record = records.get(i);
            log.info("接收Contract-product>>topic={},value={},size={}", record.topic(), record.value(), records.size());
            String coinsStr = record.value();
            List<ContractCoin> coins = JSON.parseArray(coinsStr, ContractCoin.class);
            if (coins != null && coins.size() > 0) {
                coins.stream().forEach(coin -> {
                    if (coin.getRobotStatus() == 1) {
                        if(RobotOrderRunnable.robotStatusMap.get(coin.getSymbol())!=1) {
                            map.put(coin.getSymbol(), coin.getRobotStatus());
                            contractService.robotAddOrder(coin.getSymbol());
                        }
                    }else{
                        map.put(coin.getSymbol(), coin.getRobotStatus());
                    }
                });
            }
        }
        RobotOrderRunnable.robotStatusMap = map;
    }
}
